package com.atguigu.chapter11;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Json;
import org.apache.flink.table.descriptors.Kafka;
import org.apache.flink.table.descriptors.Schema;

/**
 * @ClassName: Flink02_Table_Source_File
 * @Description:
 * @Author: kele
 * @Date: 2021/4/12 13:59
 *
 * 流程
 *      1、根据流环境，创建表环境
 *      2、创建表的模式，确定各个字段的模式Schema
 *      3、连接一个表,new kafka ,创建一个临时表
 *          --确定表的字段的名称，类型
 *          --确定字段之间的分隔符
 *          --确定表的名称
 *      4、通过临时表获取一张表
 *      5、对表进行操作
 **/
public class Flink03_Table_Source_Kafka {

    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        Schema schema = new Schema()
                            .field("id", DataTypes.STRING())
                            .field("ts",DataTypes.BIGINT())
                            .field("vc",DataTypes.INT());

        //连接，获取一个临时表
        tenv.connect(
                //从kafka读取数据，作为一个消费者
                new Kafka()
                        //从kafka读取文件的配置
                        .version("universal")
                        .property("bootstrap.servers", "hadoop162:9092")
                        .property("group.id","kafkadata")
                        .startFromLatest()  //从最新的开始读
                        .topic("senion")
        )
            .withSchema(schema)
            // .withFormat(new Csv().fieldDelimiter(','))   //从csv中读取数据
            .withFormat(new Json())    //从json字符串中读取数据
            .createTemporaryTable("temp");

        Table tab = tenv.from("temp");

        tab.execute().print();


    }



}
